package com.katze.tools.hadoop.config;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@ConditionalOnProperty(value = "tools.hadoop.enable", havingValue = "true")
public class HadoopConfiguration {

    private static final Logger log = LoggerFactory.getLogger(HadoopConfiguration.class);

    private static final String[] resources = {"hadoop-default.xml", "hadoop-site.xml", "core-default.xml", "core-site.xml", "hdfs-default.xml", "hdfs-site.xml"};

    @Bean
    public Configuration configuration() {
        Configuration configuration = new Configuration();
        for (String resource : resources) {
            Resource item = new PathMatchingResourcePatternResolver().getResource("classpath:".concat(resource));
            if (item.exists()) {
                try {
                    configuration.addResource(item.getURL());
                    log.info("{}加载成功", resource);
                } catch (Exception e) {
                    log.error("{}加载失败", resource, e);
                }
            }
        }

        String user = configuration.get("dfs.user");
        if(user != null && !user.isEmpty()){
            System.setProperty("HADOOP_USER_NAME", user);
        }
        //Kerberos认证
        if (Boolean.parseBoolean(configuration.get("hadoop.security.authorization")) && "kerberos".equalsIgnoreCase(configuration.get("hadoop.security.authentication"))){
            authenticationKerberos(configuration);
        }
        return configuration;
    }

    @Bean(initMethod = "initialization", destroyMethod = "close")
    public HadoopManager hadoopManager(Configuration configuration) {
        return new HadoopManager(configuration);
    }

    private static void authenticationKerberos(Configuration configuration){
        log.info("开始kerberos认证");
//        isKerberos = true;
//        checkDate = StringUtils.isNotBlank(configuration.get("kerberos.checkDate")) ? Integer.parseInt(configuration.get("kerberos.checkDate")) : 60;
//        kerberosTime = System.currentTimeMillis();
        System.setProperty("java.security.krb5.conf", configuration.get("java.security.krb5.conf"));
        System.setProperty("sun.security.krb5.debug",configuration.get("sun.security.krb5.debug"));
        /*System.setProperty("java.security.krb5.realm",conf.get("java.security.krb5.realm"));
        System.setProperty("java.security.krb5.kdc", conf.get("java.security.krb5.kdc"));*/
        UserGroupInformation.setConfiguration(configuration);
        try {
            UserGroupInformation.loginUserFromKeytab(configuration.get("custom.kerberos.user"),configuration.get("custom.kerberos.keytab"));
            log.info("Kerberos认证成功");
        } catch (IOException e) {
            log.error("Kerberos authentication fail", e);
        }
    }
}
